package com.xujianlong.combat;

import com.xujianlong.bean.PageViewCount;
import com.xujianlong.bean.UserBehavior;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;


import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Iterator;
import java.util.Random;

public class Flink02_Project_Project_PV2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        DataStreamSource<String> source = env.readTextFile("input/UserBehavior.csv");

        SingleOutputStreamOperator<UserBehavior> userBehaviorWithWMDS = source.map(data -> {
            String[] split = data.split(",");
            UserBehavior userBehavior = new UserBehavior(Long.parseLong(split[0]),
                    Long.parseLong(split[1]),
                    Integer.parseInt(split[2]),
                    split[3],
                    Long.parseLong(split[4]));
            return userBehavior;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(3)).withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
            @Override
            public long extractTimestamp(UserBehavior userBehavior, long l) {
                return userBehavior.getTimestamp() * 1000;
            }
        }));
        SingleOutputStreamOperator<Tuple2<String, Integer>> pvDS = userBehaviorWithWMDS.filter(data -> "pv".equals(data.getBehavior())).map(new MapFunction<UserBehavior, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(UserBehavior userBehavior) throws Exception {
                //将key+随机数
                return Tuple2.of("PV"+new Random().nextInt(8), 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStreamDS = pvDS.keyBy(data -> data.f0);

        SingleOutputStreamOperator<PageViewCount> aggregateDS = keyedStreamDS.window(TumblingEventTimeWindows.of(Time.hours(1))).aggregate(new PageViewAggFunc(), new PageViewWindowFunc());
        KeyedStream<PageViewCount, String> pageViewCountStringKeyedStream = aggregateDS.keyBy(data -> data.getTime());
        //这样每个并行度会输出一条数据
        //SingleOutputStreamOperator<PageViewCount> result = pageViewCountStringKeyedStream.sum("count");
        SingleOutputStreamOperator<PageViewCount> result = pageViewCountStringKeyedStream.process(new PageViewProcessFunc());
        result.print();

        env.execute();

    }

    public static class PageViewAggFunc implements AggregateFunction<Tuple2<String, Integer>,Integer,Integer> {

        @Override
        public Integer createAccumulator() {
            return 0;
        }

        @Override
        public Integer add(Tuple2<String, Integer> stringIntegerTuple2, Integer integer) {
            return integer+1;
        }

        @Override
        public Integer getResult(Integer integer) {
            return integer;
        }

        @Override
        public Integer merge(Integer integer, Integer acc1) {
            return integer+acc1;
        }
    }
    //in,out,key,windowType
    public static class PageViewWindowFunc implements WindowFunction<Integer, PageViewCount,String, TimeWindow>{

        @Override
        public void apply(String key, TimeWindow window, Iterable<Integer> input, Collector<PageViewCount> out) throws Exception {
            String timestamp = new Timestamp(window.getEnd()).toString();
            Integer count = input.iterator().next();
            out.collect(new PageViewCount("PV",timestamp,count));
        }
    }

    public static class PageViewProcessFunc extends KeyedProcessFunction<String,PageViewCount,PageViewCount>{
        //怎么知道它数据收集完了呢？ 窗口结束时间  定时器窗口关闭后，1s

        private ListState<PageViewCount> listState;
        @Override
        public void open(Configuration parameters) throws Exception {
            listState=getRuntimeContext().getListState(new ListStateDescriptor<PageViewCount>
                    ("list-state",PageViewCount.class));
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<PageViewCount> out) throws Exception {
            //取出状态中的数据，遍历累加
            Iterable<PageViewCount> pageViewCounts = listState.get();
            Iterator<PageViewCount> iterator = pageViewCounts.iterator();

            Integer count=0;
            while (iterator.hasNext()){
                count+=iterator.next().getCount();
            }
            //此时的count累加了多个并行度的数据
            out.collect(new PageViewCount("PV",new Timestamp(timestamp-1).toString(),count));

            listState.clear();
        }

        @Override
        public void processElement(PageViewCount value, Context ctx, Collector<PageViewCount> out) throws Exception {
            //将数据放入状态
            listState.add(value);
            //注册定时器
            String time = value.getTime();
            long ts = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(time).getTime();
            ctx.timerService().registerEventTimeTimer(ts+1);
        }
    }
}
